/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package cn.edu.thu.laud.objectstore;

import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.ColumnOrSuperColumn;
import org.apache.cassandra.thrift.ColumnParent;
import org.apache.cassandra.thrift.InvalidRequestException;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.thrift.Cassandra.Client;
import org.apache.thrift.TException;
import org.datanucleus.ClassLoaderResolver;
import org.datanucleus.OMFContext;
import org.datanucleus.ObjectManager;
import org.datanucleus.PersistenceConfiguration;
import org.datanucleus.exceptions.NucleusDataStoreException;
import org.datanucleus.metadata.AbstractClassMetaData;
import org.datanucleus.metadata.InheritanceStrategy;
import org.datanucleus.metadata.MetaDataListener;
import org.datanucleus.store.AbstractStoreManager;

import cn.edu.thu.laud.utils.CassandraUtils;
import cn.edu.thu.laud.utils.MetaDataUtils;
import cn.edu.thu.laud.utils.serde.ConverterContext;
import cn.edu.thu.laud.utils.serde.JavaSerializer;
import cn.edu.thu.laud.utils.serde.Serializer;

/**
 * 
 * @author zhuoan
 *
 */
public class CassandraStoreManager extends AbstractStoreManager{

	private boolean autoCreateSchema = false;
	private boolean autoCreateTables = false;
	
	private ObjectManager om;
	private MetaDataListener metadataListener;
	private CassandraConnectionInfo connectionInfo;
	private ConverterContext converterContext;
	
    private int poolTimeBetweenEvictionRunsMillis; 
    private int poolMinEvictableIdleTimeMillis;
    
    public static final String EVICTION_IDLE = "datanucleus.connectionPool.minEvictableIdleTimeMillis";
    public static final String EVICTION_TIME = "datanucleus.connectionPool.timeBetweenEvictionRunsMillis";
    public static final String CREATE_TABLES = "datanucleus.autoCreateTables";
    public static final String CREATE_SCHEMA = "datanucleus.autoCreateSchema";
    public static final String SERIALIZER = "cn.edu.thu.laud.utils.serde.serializer";
    public static final String BYTEMAPPER = "cn.edu.thu.laud.utils.serde.bytemapper";
    public static final String CHECKSLEEPTIME = "cn.edu.thu.laud.utils.serde.checksleep";
    
	public CassandraStoreManager(ClassLoaderResolver clr,
			OMFContext omfContext) {
		
		super("cassandra",clr, omfContext);
	   
		PersistenceConfiguration conf = omfContext.getPersistenceConfiguration();
		
		String serializerClass = conf
				.getStringProperty(SERIALIZER);

		Serializer serializer = null;
		
		if (serializerClass == null) {
			serializer = new JavaSerializer();
		} else {
			try {
				serializer = (Serializer) Class.forName(serializerClass)
						.newInstance();
			} catch (Exception e) {
				throw new NucleusDataStoreException(String.format(
						"Could not create serializer for class name %s",
						serializerClass));
			}
		}
		
		String byteMapperFile = conf.getStringProperty(BYTEMAPPER);
		
		converterContext = new ConverterContext(byteMapperFile, serializer, getOMFContext().getTypeManager(), getApiAdapter());
		
		autoCreateSchema = conf
				.getBooleanProperty(CREATE_SCHEMA);
		
		
		if(connectionInfo==null){
			connectionInfo = new CassandraConnectionInfo(conf.getStringProperty("datanucleus.ConnectionURL"), conf.getLongProperty(CHECKSLEEPTIME));
		}
		

		if (autoCreateSchema) {
			autoCreateTables = true;
			CassandraUtils.createSchema(connectionInfo);

		} else {
			autoCreateTables = conf
					.getBooleanProperty(CREATE_TABLES);
		}
		
		if (autoCreateTables) {			
			metadataListener = new CassandraMetaDataListener(this, autoCreateTables);
		    omfContext.getMetaDataManager().registerListener(metadataListener);
		}
		
		
        poolTimeBetweenEvictionRunsMillis = conf.getIntProperty(EVICTION_TIME);
        if (poolTimeBetweenEvictionRunsMillis == 0)
        {
            poolTimeBetweenEvictionRunsMillis = 15 * 1000; // default, 15 secs
        }
         
        // how long may a connection sit idle in the pool before it may be evicted
        poolMinEvictableIdleTimeMillis = conf.getIntProperty(EVICTION_IDLE);
        if (poolMinEvictableIdleTimeMillis == 0)
        {
            poolMinEvictableIdleTimeMillis = 30 * 1000; // default, 30 secs
        }
        		
		this.persistenceHandler2 = new CassandraPersistenceHandler(this);
		
		logConfiguration();
	}

	@Override
	protected void registerConnectionFactory() {
		super.registerConnectionFactory();
		this.connectionMgr.disableConnectionPool();
	}
	

	/**
	 * Accessor for the supported options in string form
	 */
	@SuppressWarnings({ "unchecked", "rawtypes" })
	public Collection getSupportedOptions() {
		Set set = new HashSet();
		set.add("ApplicationIdentity");
		set.add("TransactionIsolationLevel.read-committed");
		// could happen if writing to "one" or reading from "one" node
		set.add("TransactionIsolationLevel.read-uncommitted");
		return set;
	}

	public boolean isAutoCreateTables() {
		return autoCreateTables;
	}
	
	public boolean isAutoCreateSchema() {
		return autoCreateSchema;
	}
	
	public ObjectManager getObjectManager(){
		return om;
	}
	
	public void setObjectManager(ObjectManager om){
		
		this.om = om;
	}

	@Override
	public void close() {
		//omfContext.getMetaDataManager().deregisterListener(metadataListener);
		super.close();
	}
	
    public CassandraConnectionInfo getConnectionInfo() {
    	if(connectionInfo==null){
    		PersistenceConfiguration conf = omfContext.getPersistenceConfiguration();
    		connectionInfo =  new CassandraConnectionInfo(conf.getStringProperty("datanucleus.ConnectionURL"), conf.getLongProperty(CHECKSLEEPTIME));	
    	}	
		return connectionInfo;
	}

	public int getPoolMinEvictableIdleTimeMillis()
    {
        return poolMinEvictableIdleTimeMillis;
    }
    
    public int getPoolTimeBetweenEvictionRunsMillis()
    {
        return poolTimeBetweenEvictionRunsMillis;
    }
    
       

	/**
	 * @return the byteConverterContext
	 */
	public ConverterContext getConverterContext() {
		return converterContext;
	}
	
	/*
	 * (non-Javadoc)
	 * 
	 * @see
	 * org.datanucleus.store.AbstractStoreManager#getClassNameForObjectID(java
	 * .lang.Object, org.datanucleus.ClassLoaderResolver,
	 * org.datanucleus.store.ExecutionContext)
	 */
	@Override
	public String getClassNameForObjectID(Object id, ClassLoaderResolver clr,
			ObjectManager om) {

		String pcClassName = super.getClassNameForObjectID(id, clr, om);

		AbstractClassMetaData metaData = om.getMetaDataManager()
				.getMetaDataForClass(pcClassName, clr);

		SlicePredicate descriminator = MetaDataUtils.getDescriminatorColumn(metaData);

		// We only support discriminator. Even in a subclass per table scheme
		// for clarity of the columns within Cassandra.
		if (descriminator == null) {
			return pcClassName;
		}

		ByteBuffer key = converterContext.getRowKeyForId(id);
		Client client = (Client)((CassandraManagedConnection)this.getConnection(om)).getConnection();
		try {
			client.set_keyspace(this.connectionInfo.getKeyspace());
		} catch (Exception e) {
			// TODO Auto-generated catch block
			throw new NucleusDataStoreException(e.getMessage(), e);
		}
		return findObject(client, key, metaData, clr, om, id);

	}

	private String findObject(Cassandra.Client client, ByteBuffer key, AbstractClassMetaData metaData,
			ClassLoaderResolver clr, ObjectManager om, Object id) {

		// if we have a discriminator, fetch the discriminator column only
		// and see if it's equal
		// to the class provided by the op

		List<ColumnOrSuperColumn> columns = null;
		
		try {
			columns = client.get_slice(key, new ColumnParent(MetaDataUtils.getColumnFamily(metaData)), MetaDataUtils.getDescriminatorColumn(metaData), CassandraConsistency.get());

		} catch (Exception e) {
			throw new NucleusDataStoreException(e.getMessage(), e);
		}

		// what do we do if no descriminator is found and one should be
		// present?
		if (columns == null || columns.size() != 1) {

			// now check if we have subclasses from the given metaData, if we do
			// recurse to a child class and search for the object
			String[] decendents = om.getMetaDataManager()
					.getSubclassesForClass(metaData.getFullClassName(), true);

			// it has decendents, only recurse to them if their inheritance
			// strategy is a new table
			if (decendents == null || decendents.length == 0) {
				return null;
			}

			AbstractClassMetaData decendentMetaData = null;

			for (String decendent : decendents) {
				decendentMetaData = om.getMetaDataManager()
						.getMetaDataForClass(decendent, clr);

				InheritanceStrategy strategy = decendentMetaData
						.getInheritanceMetaData().getStrategy();

				// either the subclass has it's own table, or one if it's
				// children may, recurse to find the object
				if (InheritanceStrategy.NEW_TABLE.equals(strategy)
						|| InheritanceStrategy.SUBCLASS_TABLE.equals(strategy)) {
					String result = findObject(client, key, decendentMetaData, clr, om,
							id);

					// we found a subclass with the descriminator stored, return
					// it
					if (result != null) {
						return result;
					}
				}
			}

			// nothing found in this class or it's children return null
			return null;

		}

		String descriminatorValue = converterContext.getString(columns.get(0).getColumn().value);

		String className = org.datanucleus.metadata.MetaDataUtils
				.getClassNameFromDiscriminatorValue(descriminatorValue,
						metaData.getDiscriminatorMetaData(), om);

		// now recursively load the search for our class
		return className;
	}
    
}
